"""
Case Type   : 基础功能
Case Name   : 订阅发布创建1千个表
Create At   : 2022/1/28
Owner       : opentestcase026
Description :
    1.在两个集群创建表相同结构1000个表
    2.创建发布端
    3.创建订阅
    4.发布端每张表插入100条数据
    5.查询订阅端是否同步
    6.发布端更新数据
    7.查询订阅端是否同步
    8.发布端删除数据
    9.查询订阅端是否同步
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.数据同步
    6.更新成功
    7.数据同步
    8.删除成功
    9.数据同步
History     :
    Modified by opentestcase026 2022/3/15:研发代码变更，gsql -c "" -f *.sql 无法执行，
    必须删除-c ""
    Modified by lopentestcase027 2022/8/4:优化用例,先建表后建订阅 设置copy_data=false
    Modified by opentestcase004 2022/09/26:因为发布端有1W张表，创建订阅要2个多小时，
    ci超时1.5h会和其他用例并行，造成互相干扰，与开发沟通，减少表数目
    Modified by opentestcase013 2023/03/22:优化用例，减少用例执行时长
    Modified by opentestcase012 2023/5/5:优化用例,修改调用的env路径
    Modified by opentestcase024 2023/8/11:优化用例逻辑，使用匿名块进行大量表及数据构造
"""
import os
import time
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("----this is setup----")
        self.log.info(f"----{os.path.basename(__file__)} start----")
        self.pri_userdb_list = [
            Node(node='PrimaryDbUser'),
            Node(node='remote1_PrimaryDbUser')]
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_sub = CommonSH('remote1_PrimaryDbUser')
        self.commsh_sub.DB_ENV_PATH = macro.DB_ENV_PATH_REMOTE1
        self.commsh_sub.DB_INSTANCE_PATH = macro.DB_INSTANCE_PATH_REMOTE1
        self.com_pub = Common()
        self.tb_name = 'tb_pubsub_case124'
        self.subname = "sub_case124"
        self.pubname = "pub_case124"
        self.tb_num = 1000
        self.row_num = 100
        self.parent_path_pub = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.parent_path_sub = os.path.dirname(macro.DB_INSTANCE_PATH_REMOTE1)
        self.port = str(int(self.pri_userdb_list[0].db_port) + 1)
        self.wal_level = self.com_pub.show_param("wal_level")
        self.user_param_pub = f'-U {self.pri_userdb_list[0].db_user} ' \
            f'-W {self.pri_userdb_list[0].db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_list[1].db_user} ' \
            f'-W {self.pri_userdb_list[1].db_password}'

        cmd = f"cp {os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')} " \
            f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_list[0].sh(cmd).result()
        self.log.info(result)
        cmd = f"cp " \
            f"{os.path.join(macro.DB_INSTANCE_PATH_REMOTE1, 'pg_hba.conf')}" \
            f" {os.path.join(self.parent_path_sub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_list[1].sh(cmd).result()
        self.log.info(result)

        text = '----step:预置条件,修改pg_hba expect:成功'
        self.log.info(text)
        self.log.info("###########发布端：")
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_list[1].db_user} '
            f'{self.pri_userdb_list[1].db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.restart_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in result or 'Degrade' in result
        self.assertTrue(flg, '执行失败:' + text)
        self.log.info("###########订阅端：")
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_list[0].db_user} '
            f'{self.pri_userdb_list[0].db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)

        self.create_cmd = f"declare\n" \
            f"v_strtable text;\n" \
            f"v_strsql text;\n" \
            f"begin\n" \
            f"for i in 1..{self.tb_num} loop\n" \
            f"v_strtable := concat('{self.tb_name}_', i);\n" \
            f"v_strsql = 'create table '||v_strtable||" \
            f"'(iggggggggggggggggggggggggggggg1 int primary key, " \
            f"iggggggggggggggggggggggggggggggg2 point, " \
            f"iggggggggggggggggggggg3 lseg, igggggggggggggggggggggg4 box, " \
            f"igggggggggggggggggggggggggggggggggggggggggggggggggggggg5 path," \
            f"igggggggggggggggggggggggggggggggggggggggggggggggggg6 path," \
            f"i7 polygon,i8 circle, i9 bit(3),i10 bit varying(5))';\n" \
            f"execute v_strsql; \n" \
            f"end loop;\n" \
            f"end;"
        self.insert_sql = f"declare\n" \
            f"v_strtable text;\n" \
            f"v_strsql text;\n" \
            f"begin\n" \
            f"for i in 1..{self.tb_num} loop\n" \
            f"for j in 1..{self.row_num} loop\n" \
            f"v_strtable := concat('{self.tb_name}_', i);\n" \
            f"v_strsql ='insert into '||v_strtable||' values('||j||', " \
            f"point''2,3'',lseg''[(1,2),(2,2)]'',box''(2,4),(4,4)''," \
            f"path''((1,2),(1,3),(3,2),(3,3))''," \
            f"path''[(1,1),(1,2),(1,3)]'',polygon''(1,2),(1,3),(3,2),(3,3)''," \
            f"circle''4,4,2'',b''101'',b''00'')';\n" \
            f"execute v_strsql; \n" \
            f"end loop;\n" \
            f"end loop;\n" \
            f"end;"
        self.update_sql = f"declare\n" \
            f"v_strtable text;\n" \
            f"v_strsql text;\n" \
            f"begin\n" \
            f"for i in 1..{self.tb_num} loop\n" \
            f"v_strtable := concat('{self.tb_name}_', i);\n" \
            f"v_strsql ='update '||v_strtable||' " \
            f"set iggggggggggggggggggggggggggggggg2 = point''3,4''';\n" \
            f"execute v_strsql; \n" \
            f"end loop;\n" \
            f"end;"
        self.delete_sql = f"declare\n" \
            f"v_strtable text;\n" \
            f"v_strsql text;\n" \
            f"begin\n" \
            f"for i in 1..{self.tb_num} loop\n" \
            f"v_strtable := concat('{self.tb_name}_', i);\n" \
            f"v_strsql ='delete from '||v_strtable;\n" \
            f"execute v_strsql; \n" \
            f"end loop;\n" \
            f"end;"
        self.drop_sql = f"declare\n" \
            f"v_strtable text;\n" \
            f"v_strsql text;\n" \
            f"begin\n" \
            f"for i in 1..{self.tb_num} loop\n" \
            f"v_strtable := concat('{self.tb_name}_', i);\n" \
            f"v_strsql ='drop table '||v_strtable;\n" \
            f"execute v_strsql; \n" \
            f"end loop;\n" \
            f"end;"

    def test_pubsub(self):
        text = f'----step1:在两个集群创建表相同结构{self.tb_num}个表 expect:成功----'
        self.log.info(text)
        result = self.commsh_pub.execut_db_sql(self.create_cmd,
                                               self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.CREATE_ANONYMOUS_BLOCK_SUCCESS_MSG,
                      result, '执行失败:' + text)
        result = self.commsh_sub.execut_db_sql(self.create_cmd,
                                               self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.CREATE_ANONYMOUS_BLOCK_SUCCESS_MSG,
                      result, '执行失败:' + text)

        text = '----step2:创建发布端 expect:成功----'
        self.log.info(text)
        sql = f"create publication {self.pubname} for all tables;"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.create_pub_succ_msg, result,
                      '执行失败:' + text)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                         '执行失败:' + text)

        text = '----step3:创建订阅 expect:成功----'
        self.log.info(text)
        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn('', result, '执行失败:' + text)
        sql = f"create subscription {self.subname} connection " \
            f"'host={self.pri_userdb_list[0].db_host} " \
            f"port={self.port} " \
            f"user={self.pri_userdb_list[0].db_user} " \
            f"dbname={self.pri_userdb_list[0].db_name} " \
            f"password={self.pri_userdb_list[0].ssh_password}' " \
            f"publication {self.pubname};"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.create_sub_succ_msg,
                      result, '执行失败:' + text)

        self.log.info("----等待订阅复制状态为r----")
        check_r_sql = "select * from pg_subscription_rel;"
        for i in range(500):
            result = self.commsh_sub.execut_db_sql(check_r_sql,
                                                   self.user_param_sub,
                                                   None,
                                                   macro.DB_ENV_PATH_REMOTE1)
            _, cols = self.com_pub.parse_sql_query_res(result, False)
            srsubstate = cols[2]
            if set(srsubstate) == {'r'}:
                self.log.info('第' + str(i) + 's复制状态正常')
                self.log.info(result)
                break
            time.sleep(1)
        self.log.info("----等待订阅端数据同步，超时时间为60s----")
        self.commsh_pub.check_sub_consistency(self.subname,
                                              self.user_param_pub, 60)

        text = f'----step4:发布端每张表插入{self.row_num}条数据 expect:成功----'
        self.log.info(text)
        result = self.commsh_pub.execut_db_sql(self.insert_sql,
                                               self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.CREATE_ANONYMOUS_BLOCK_SUCCESS_MSG,
                      result, '执行失败:' + text)

        self.log.info("----等待订阅端数据同步，超时时间为60s----")
        self.commsh_pub.check_sub_consistency(self.subname,
                                              self.user_param_pub, 60)
        self.log.info("----等待订阅主备端数据同步，超时时间为60s----")
        self.commsh_sub.check_location_consistency(
            'primary', 3, 60, macro.DB_ENV_PATH_REMOTE1,
            macro.DB_INSTANCE_PATH_REMOTE1)

        text = "----step5:查询订阅端是否同步 expect:数据同步----"
        self.log.info(text)
        select_cmd = ''
        for i in range(1, self.tb_num + 1):
            select_cmd = select_cmd + f"select count(*) from " \
                f"{self.tb_name}_{i};"
        result1 = self.commsh_pub.execut_db_sql(select_cmd,
                                                self.user_param_pub, None,
                                                macro.DB_ENV_PATH)
        result2 = self.commsh_sub.execut_db_sql(select_cmd,
                                                self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.assertEqual(result1, result2, '执行失败:' + text)

        text = "----step6:发布端更新数据 expect:更新成功----"
        self.log.info(text)
        result = self.commsh_pub.execut_db_sql(self.update_sql,
                                               self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.CREATE_ANONYMOUS_BLOCK_SUCCESS_MSG,
                      result, '执行失败:' + text)

        self.log.info("----等待订阅端数据同步，超时时间为60s----")
        self.commsh_pub.check_sub_consistency(self.subname,
                                              self.user_param_pub, 60)
        self.log.info("----等待订阅主备端数据同步，超时时间为60s----")
        self.commsh_sub.check_location_consistency(
            'primary', 3, 60, macro.DB_ENV_PATH_REMOTE1,
            macro.DB_INSTANCE_PATH_REMOTE1)

        text = "----step7:查询订阅端是否同步 expect:数据同步----"
        self.log.info(text)
        result1 = self.commsh_pub.execut_db_sql(select_cmd,
                                                self.user_param_pub, None,
                                                macro.DB_ENV_PATH)
        result2 = self.commsh_sub.execut_db_sql(select_cmd,
                                                self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.assertEqual(result1, result2, '执行失败:' + text)

        text = "----查询其中一张表订阅端数据是否正确 expect:数据同步----"
        self.log.info(text)
        selectd_cm1 = f"select * from {self.tb_name}_{900};"
        result1 = self.commsh_pub.execut_db_sql(selectd_cm1,
                                                self.user_param_pub, None,
                                                macro.DB_ENV_PATH)
        result2 = self.commsh_sub.execut_db_sql(selectd_cm1,
                                                self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.assertEqual(result1, result2, '执行失败:' + text)

        text = "----step8:发布端删除数据 expect:删除成功----"
        self.log.info(text)
        result = self.commsh_pub.execut_db_sql(self.delete_sql,
                                               self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.CREATE_ANONYMOUS_BLOCK_SUCCESS_MSG,
                      result, '执行失败:' + text)

        self.log.info("----等待订阅端数据同步，超时时间为60s----")
        self.commsh_pub.check_sub_consistency(self.subname,
                                              self.user_param_pub, 60)
        self.log.info("----等待订阅主备端数据同步，超时时间为60s----")
        self.commsh_sub.check_location_consistency(
            'primary', 3, 60, macro.DB_ENV_PATH_REMOTE1,
            macro.DB_INSTANCE_PATH_REMOTE1)

        text = "----step9:查询订阅端是否同步 expect:数据同步----"
        self.log.info(text)
        result1 = self.commsh_pub.execut_db_sql(select_cmd,
                                                self.user_param_pub, None,
                                                macro.DB_ENV_PATH)
        result2 = self.commsh_sub.execut_db_sql(select_cmd,
                                                self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.assertEqual(result1, result2, '执行失败:' + text)

    def tearDown(self):
        self.log.info('----this is tearDown----')
        text = '----清理环境----'
        self.log.info(text)
        text1 = '----删除订阅----'
        self.log.info(text1)
        sql = f"drop subscription  {self.subname};"
        drop_sub_result = self.commsh_sub.execut_db_sql(
            sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_result)
        text2 = '----删除发布----'
        self.log.info(text2)
        sql = f"drop publication if exists {self.pubname};"
        drop_pub_result = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result)
        text3 = '----清理发布端表----'
        self.log.info(text3)
        drop_pub_tb = self.commsh_pub.execut_db_sql(self.drop_sql,
                                                    self.user_param_pub)
        self.log.info(drop_pub_tb)
        text4 = '----清理订阅端表----'
        self.log.info(text4)
        drop_sub_tb = self.commsh_sub.execut_db_sql(self.drop_sql,
                                                    self.user_param_sub, None,
                                                    macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_tb)
        text5 = '----还原参数及配置文件----'
        self.log.info(text5)
        cmd = f"mv {os.path.join(self.parent_path_pub, 'pg_hba.conf')} " \
            f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')};"
        self.log.info(cmd)
        mv_pub = self.pri_userdb_list[0].sh(cmd).result()
        self.log.info(mv_pub)
        cmd = f"mv {os.path.join(self.parent_path_sub, 'pg_hba.conf')} " \
            f"{os.path.join(macro.DB_INSTANCE_PATH_REMOTE1, 'pg_hba.conf')};"
        self.log.info(cmd)
        mv_sub = self.pri_userdb_list[1].sh(cmd).result()
        self.log.info(mv_sub)
        set_result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        restart_pub = self.commsh_pub.restart_db_cluster(False)
        restart_sub = self.commsh_sub.restart_db_cluster(
            False, macro.DB_ENV_PATH_REMOTE1)

        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                      '执行失败' + text1)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                      '执行失败' + text2)
        self.assertIn(self.constant.CREATE_ANONYMOUS_BLOCK_SUCCESS_MSG,
                      drop_pub_tb, '执行失败:' + text3)
        self.assertIn(self.constant.CREATE_ANONYMOUS_BLOCK_SUCCESS_MSG,
                      drop_sub_tb, '执行失败:' + text4)
        self.assertEqual(mv_pub, '', '执行失败:' + text5)
        self.assertEqual(mv_sub, '', '执行失败:' + text5)
        self.assertTrue(set_result, '执行失败:' + text5)
        self.assertTrue(restart_pub, '执行失败:' + text5)
        self.assertTrue(restart_sub, '执行失败:' + text5)

        self.log.info(f"----{os.path.basename(__file__)} end----")
